package org.activequant.tradesystem.ib;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.marketscanner.InstrumentSpecificationDetails;
import org.activequant.core.domainmodel.marketscanner.MarketScan;
import org.activequant.core.domainmodel.marketscanner.MarketScanRow;
import org.activequant.core.domainmodel.marketscanner.ScanCriteria;
import org.activequant.core.types.TimeStamp;
import org.activequant.data.retrieval.IMarketScanSubscription;
import org.activequant.tradesystem.types.Mode;
import org.activequant.util.pattern.events.IEventListener;
import org.apache.log4j.Logger;

import com.ib.algokit.InstrumentAnalyzerState;
import com.ib.algokit.RunMe;
import com.ib.algokit.TraderProxy;
import com.ib.algokit.marketscan.IMarketScanEventProcessingStrategy;
import com.ib.algokit.marketscan.MarketScanChangeEventDetectionStrategy;
/**
 * IB is capable of running up to 10 MarketScanWorker at the same time, and then we have to disconnect/reconnect to be able to use another ones. 
 * This class handles that logic and defines the nested MarketScanWorker class.<br/>
 * Holds the following associated variables:
 * <ul>
 * <li>totalAttemptsPerSession(int)</li>
 * <li>availableScanners(Map&lt;ScanCriteria,MarketScanWorker&gt;)</li>
 * <li>unAvailableScanners(Map&lt;ScanCriteria,MarketScanWorker&gt;)</li>
 * <li>activeScanners(Queue&lt;MarketScanWorker&gt;)</li>
 * <li>ema(TraderProxy)</li>
 * </ul>
 * @author Dimitar
 *
 */
public class MarketScannerFactory {
	protected final static Logger log = Logger.getLogger(MarketScannerFactory.class);
	/**
	 * public final static int IB_SIMULTANEOUS_SCAN_LIMIT=10;<br/>
	 * IB allows up to 10 marketscanner subscribes, regardless if you cancel them. After that we have to reconnect
	 */
	public final static int IB_SIMULTANEOUS_SCAN_LIMIT=10;
	/**
	 * private int totalAttemptsPerSession=0;<br/>
	 * this counts the total number of attempts for this session
	 */
	private int totalAttemptsPerSession=0;
	/**
	 * private Map&lt;ScanCriteria,MarketScanWorker&gt; availableScanners=new ConcurrentHashMap&lt;ScanCriteria,MarketScanWorker&gt;();<br/>
	 * what scanners we know work right now, not necessarily activated
	 */
	private Map<ScanCriteria,MarketScanWorker> availableScanners=new ConcurrentHashMap<ScanCriteria,MarketScanWorker>();
	/**
	 * private Map&lt;ScanCriteria,MarketScanWorker&gt; unAvailableScanners=new ConcurrentHashMap&lt;ScanCriteria,MarketScanWorker&gt;();<br/>
	 * scanners that didn't respond (timed out)
	 */
	private Map<ScanCriteria,MarketScanWorker> unAvailableScanners=new ConcurrentHashMap<ScanCriteria,MarketScanWorker>();
	/**
	 * private Queue&lt;MarketScanWorker&gt; activeScanners = new ConcurrentLinkedQueue&lt;MarketScanWorker&gt;();
	 */
	private Queue<MarketScanWorker> activeScanners = new ConcurrentLinkedQueue<MarketScanWorker>();
	/**
	 * private final EMA ema;
	 */
	private final TraderProxy ema;
	/**
	 * private static Map&lt;InstrumentSpecificationDetails,InstrumentSpecificationDetails&gt; scannerSpecDetailsCache=new HashMap&lt;InstrumentSpecificationDetails,InstrumentSpecificationDetails&gt;();<br/>
	 * this is supposed to keep database looked up specification values for each previously received instrument from the scanner so i don't need to lookup them up in the database every time
	 */
	private static Map<InstrumentSpecificationDetails,InstrumentSpecificationDetails> scannerSpecDetailsCache=new HashMap<InstrumentSpecificationDetails,InstrumentSpecificationDetails>();
	/**
	 * private Queue&lt;InstrumentSpecificationDetails&gt; specsPendingFetch = new ConcurrentLinkedQueue&lt;InstrumentSpecificationDetails&gt;();<br/>
	 * this holds the instrument spec details that need to be fetched
	 */
	private static Queue<InstrumentSpecificationDetails> specsPendingFetch = new ConcurrentLinkedQueue<InstrumentSpecificationDetails>();
	/**
	 * private static LinkedBlockingDeque&lt;MarketScan&gt; marketScanPendingFetch = new LinkedBlockingDeque&lt;MarketScan&gt;();<br/>
	 * this holds the marketScans pending fetch
	 */
	private static LinkedBlockingDeque<MarketScan> marketScanPendingFetch = new LinkedBlockingDeque<MarketScan>();
	/**
	 * private static AtomicReference&lt;InstrumentSpecificationDetails&gt; specDetailsBag = new AtomicReference&lt;InstrumentSpecificationDetails&gt;();<br/>
	 * this holds the currently fetched spec details (only 1 call at a time is allowed)
	 */
	private static AtomicReference<InstrumentSpecificationDetails> specDetailsBag = new AtomicReference<InstrumentSpecificationDetails>();
	/**
	 * private static AtomicReference&lt;MarketScan&gt; currentMarketScan = new AtomicReference&lt;MarketScan&gt;();<br/>
	 * this is the marketScan I am currently working to save (fetch each instrument specification as needed)
	 */
//	private static AtomicReference<MarketScan> currentMarketScan = new AtomicReference<MarketScan>();
	/**
	 * public final static boolean useDatabaseForRealTimeMarketScans=true;
	 */
	public final static boolean useDatabaseForRealTimeMarketScans=true;
	/**
	 * private final InstrumentAnalyzerState state;
	 */
	private final InstrumentAnalyzerState state;
	/**
	 * constructs a MarketScannerFactory
	 * @param marketScanSubscriptionSource
	 */
	public MarketScannerFactory(TraderProxy ema){
		this.ema=ema;		
		this.state=new InstrumentAnalyzerState();
//		if(RunMe.isRealRun&&useDatabaseForRealTimeMarketScans){
			new Thread(new Runnable() {
				public void run() {
					saveMarketScans();
				}
			}).start();
//		}
	}
	
	public MarketScanWorker getScanner(ScanCriteria scanCriteria){
		return getScanner(scanCriteria, false);
	}
	/**
	 * <strong>1.</strong> First it does a few checks for efficiency:<br/>
	 * - Checks if the given scanCriteria(ScanCriteria) was previously seen as unavailable, and if yes, it doesn't bother with it again, but returns null right away.<br/>
	 * - Checks if the given scanCriteria(ScanCriteria) was previously marked as available, and if yes, reuses the same MarketScanWorker and keeps adding MarketScan's to any ScanHistory already there.<br/>
	 * - Otherwise it creates/activates a new MarketScanWorker for the given scanCriteria(ScanCriteria).<br/>
	 * <strong>2.</strong> Increases the number of attempts and tries to activate the MarketScan subscription (Unfortunately IB allows up to 10 attempts regardless whether they were successful or not.<br/>
	 * (After that we have to reconnect, as canceling them doesn't help either:( ).<br/>
	 * <strong>3.</strong> Depending whether the marketScanner returned anything or not we place it into the available or unavailable scanners, and if necessary reconnect to TWS. The given permanent(boolean)
	 * determines whether it will automatically reconnect after the 10th attempt or not.
	 * @param scanCriteria
	 * @param permanent
	 * @return
	 */
	public MarketScanWorker getScanner(ScanCriteria scanCriteria, boolean permanent){
		if(unAvailableScanners.containsKey(scanCriteria)){
			log.info("scanner unavailable "+scanCriteria);
			return null;
		}
		//log.info("activeScanners.size() "+activeScanners.size()+" avScanners.size() "+availableScanners.size()+" unavScanners.size() "+unAvailableScanners.size());
		MarketScanWorker worker=availableScanners.get(scanCriteria);
		if(worker==null){
			worker=new MarketScanWorker(scanCriteria,this.ema);
		}	
		totalAttemptsPerSession++;
		try {			
			worker.activate();
		} catch (Exception e) {
			log.error(e.getMessage());
		}finally{
			if (worker.isResponding()) {
				activeScanners.offer(worker);
				availableScanners.put(scanCriteria, worker);
				if (!permanent&& totalAttemptsPerSession >= IB_SIMULTANEOUS_SCAN_LIMIT) {
					resetMarketScanSubscriptions();
				}else if (totalAttemptsPerSession >IB_SIMULTANEOUS_SCAN_LIMIT) {
					resetMarketScanSubscriptions();
				}
			} else {
				worker.cancel();
				unAvailableScanners.put(scanCriteria, worker);
			}
		}
		if (worker.isResponding()){
			return worker;
		}else{
			return null;
		}
	}
	/**
	 * Cancels and removes all currently active MarketScanWorkers from the associated activeScanners(Queue&lt;MarketScanWorker&gt;), reconnects to TWS and resets the 
	 * associated totalAttemptsPerSession(int) to 0
	 */
	public void resetMarketScanSubscriptions(){
		while(!activeScanners.isEmpty()){
			MarketScanWorker marketScanWorker = activeScanners.poll();
			if(marketScanWorker!=null){
				marketScanWorker.cancel();
			}
		}	
		ema.getMarketScanSubscriptionSource().reconnect();
		totalAttemptsPerSession=0;
	}
	/**
	 * private boolean processMarketScans=false;
	 */
	private boolean processMarketScans=true;
	
	public void saveMarketScans(){
		while(true){
			try {
				MarketScan marketScan=marketScanPendingFetch.takeFirst();
				//test code
//				StringBuilder info=new StringBuilder("pre code="+marketScan.getScanCriteria().getScanCode()+"@"+marketScan.getTimeStamp().getDate()+" "+marketScan.getRows().size()+"rows");
//				List<MarketScanRow> marketScanRows= marketScan.getRows();
//				for(MarketScanRow marketScanRow : marketScanRows){
//					InstrumentSpecification instrumentSpecification = marketScanRow.getInstrumentSpecificationDetails().getInstrumentSpecification();
//					info.append(" "+instrumentSpecification.getSymbol());
//				}
//				log.info(info);
				//test code end
				
				boolean isFetchFromIBRequired=false;
				for(MarketScanRow row:marketScan.getRows()){
					InstrumentSpecificationDetails specDetails=row.getInstrumentSpecificationDetails();
					InstrumentSpecificationDetails cachedSpecDetailsDB=scannerSpecDetailsCache.get(specDetails);
					if(cachedSpecDetailsDB!=null){//if found in cache, replace with that
						row.setInstrumentSpecificationDetails(cachedSpecDetailsDB);
					}else{
						InstrumentSpecification spec = specDetails.getInstrumentSpecification();
						//otherwise, lookup from database and if there put in cache, if not request from IB, store and cache
						InstrumentSpecification specFromDB = spec.hasId()? spec: ema.getInstrumentSpecificationDao().findByExample(spec);
						InstrumentSpecificationDetails specDetailsFromDB = specDetails.hasId()?specDetails: (specFromDB!=null? (ema.getSpecificationDetailsDao().findByInstrumentSpecification(specFromDB)) : null);
						//found spec in database, now get spec details from database as well, most of the time it will be there if the spec is
						if(specFromDB!=null&&specDetailsFromDB!=null){
							//both spec and specDetails were wound in the database, cache scan - db pair
							scannerSpecDetailsCache.put(specDetails, specDetailsFromDB);
							//set db version in marketScan row
							row.setInstrumentSpecificationDetails(specDetailsFromDB);							
						}else{
							//not found in cache or database, queue up for fetching from IB
							specsPendingFetch.offer(specDetails);
							//we ll have to wait for the missing ones to get fetched
							isFetchFromIBRequired=true;
						}					
					}				
				}
				if(RunMe.mode==Mode.LIVE&&useDatabaseForRealTimeMarketScans){	
					if(isFetchFromIBRequired){
						while(!specsPendingFetch.isEmpty()){
							InstrumentSpecificationDetails specDetails=specsPendingFetch.poll();					
							requestInstrumentSpecificationDetails(specDetails);
						}
						state.waitSpecDetailsFinish();
						for(MarketScanRow row:marketScan.getRows()){
							InstrumentSpecificationDetails specDetails=row.getInstrumentSpecificationDetails();
							if(!specDetails.hasId()){
								InstrumentSpecificationDetails cachedSpecDetailsDB=scannerSpecDetailsCache.get(specDetails);
								if(cachedSpecDetailsDB!=null){
									row.setInstrumentSpecificationDetails(cachedSpecDetailsDB);
								}else{
									log.warn("[saveMarketScans] specDetails missing even after requesting it from IB");				
								}
							}
						}
					}					
					ema.getMarketScanDao().update(marketScan);
					
				}
				if(processMarketScans){
					ema.getMarketScanListener().eventFired(marketScan);
				}
				
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
	private final IEventListener<ArrayList<InstrumentSpecificationDetails>> instrumentSpecificationDetailsListener = new IEventListener<ArrayList<InstrumentSpecificationDetails>>() {
		public void eventFired(ArrayList<InstrumentSpecificationDetails> detailsList) {
			if(detailsList.size()>1){
				log.warn("[MarketScannerFactory] Got more than 1 specDetails... something is fishy");				
			}else{
				for(InstrumentSpecificationDetails details:detailsList){
					InstrumentSpecification spec=details.getInstrumentSpecification();
					//save in ddatabase
					details=ema.getSpecificationDetailsDao().update(details);
					InstrumentSpecificationDetails specDetails = specDetailsBag.getAndSet(null);
					//make sure its the right one
					assert(specDetails.getInstrumentSpecification().getSymbol().toString().equals(spec.getSymbol().toString()));	
					//put in cache
					scannerSpecDetailsCache.put(specDetails, details);					
				}
			}
			state.setSpecDetailsRequestReceived(detailsList.size()!=0);
		}
	};

	public void requestInstrumentSpecificationDetails(InstrumentSpecificationDetails details) {	
		InstrumentSpecification instrumentSpecification=details.getInstrumentSpecification();
		state.resetSpecDetailsRequestState();
		state.standInLineForSpecDetailsRequest();
		if(specDetailsBag.getAndSet(details)!=null){
			log.warn("can't request another instrument until the previous one is done.");
			return;
		}
		try {
			ema.getContractDetailsSource().requestInstrumentSpecificationDetails(instrumentSpecification, instrumentSpecificationDetailsListener);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		state.waitSpecDetailsFinish();
	}
	/**
	 * A nested static class. MarketScanWorker implements IEventListener&lt;MarketScan&gt;.
	 * Holds the following associated variables:
	 * <ul>
	 * <li>lastScanReceived(TimeStamp)</li>
	 * <li>lastAttempted(TimeStamp)</li>
	 * <li>isResponding(AtomicBoolean)</li>
	 * <li>ema(EMA)</li>
	 * <li>liveMarketScans(List&lt;MarketScan&gt;)</li>
	 * <li>previousMarketScan(MarketScan)</li>
	 * <li>scanCriteria(ScanCriteria)</li>
	 * <li>subscription(IMarketScanSubscription&lt;MarketScan&gt;)</li>
	 * <li>marketScanEventProcessingStrategy(IMarketScanEventProcessingStrategy)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	public static class MarketScanWorker implements IEventListener<MarketScan>{
		/**
		 * private final static int CACHE_SIZE=15;<br/>
		 * this limits the number of MarketScan's we keep in history for each MarketScanWorker
		 */
		private final static int CACHE_SIZE=15;
		/**
		 * private final static int MARKET_SCAN_TIMEOUT=15000000;<br/>
		 * this determines how much we wait to see if the market scanner is working(until it returns/fires the first value)
		 */
		private final static int MARKET_SCAN_TIMEOUT=15000000;
		/**
		 * private TimeStamp lastScanReceived;
		 */
		private TimeStamp lastScanReceived;
		/**
		 * private TimeStamp lastAttempted;
		 */
		private TimeStamp lastAttempted;
		/**
		 * private AtomicBoolean isResponding=new AtomicBoolean(false);
		 */
		private AtomicBoolean isResponding=new AtomicBoolean(false);
		/**
		 * private final EMA ema;
		 */
		private final TraderProxy ema;
		/**
		 * private List&lt;MarketScan&gt; liveMarketScans = new ArrayList&lt;MarketScan&gt;();
		 */
		private List<MarketScan> liveMarketScans = new ArrayList<MarketScan>();
		/**
		 * private MarketScan previousMarketScan;
		 */
		private MarketScan previousMarketScan;
		/**
		 * private ScanCriteria scanCriteria;
		 */
		private final ScanCriteria scanCriteria;
		/**
		 * private IMarketScanSubscription&lt;MarketScan&gt; subscription;
		 */
		private IMarketScanSubscription<MarketScan> subscription;
		/**
		 * private IMarketScanEventProcessingStrategy marketScanEventProcessingStrategy=new MarketScanChangeEventDetectionStrategy();
		 */
		private IMarketScanEventProcessingStrategy marketScanEventProcessingStrategy=new MarketScanChangeEventDetectionStrategy();	
		/**
		 * private boolean processMarketScanChanges=false;
		 */
		private boolean processMarketScanChanges=false;
		/**
		 * constructs a MarketScanWorker(implements IEventListener&lt;MarketScan&gt;) using the given scanCriteria(ScanCriteria) and ema(TraderProxy)
		 * to set its associated scanCriteria(ScanCriteria) and ema(TraderProxy)
		 * @param scanCriteria
		 * @param mktScanSubscriptionSource
		 */
		public MarketScanWorker(ScanCriteria scanCriteria, TraderProxy ema) {
			this.scanCriteria = scanCriteria;
			this.ema=ema;
		}
		/**
		 * Activates this MarketScanWorker, waiting up to MARKET_SCAN_TIMEOUT(15 seconds) for it to return at least one result.
		 * If it doesn't respond, it cancels it, and stops listening to it.<br/>
		 * - Sets the associated lastAttempted(TimeStamp) with the current time and registers itself to listen for any events.
		 * - Activates the associated subscription(IMarketScanSubscription&lt;MarketScan&gt;) and waits up to 15 secs to see whether it will respond. If it doesn't, it cancels it right away
		 * @throws Exception
		 */
		public void activate() throws Exception{
			lastAttempted=ema.getTimeStampService().currentTimeStamp();
			subscription=ema.getMarketScanSubscriptionSource().subscribe(scanCriteria);
			subscription.addEventListener(MarketScanWorker.this);	
			subscription.activate();
			isResponding.set(false);
			synchronized(isResponding) {
				try {
					isResponding.wait(MARKET_SCAN_TIMEOUT);
				} catch(InterruptedException ex) {
				}
			}
			if(!isResponding.get()) {
				cancel();
				log.info("timeout waiting for scanner data... deactivated scanner");
			}
		}
		/**
		 * If the associated subscription(IMarketScanSubscription&lt;MarketScan&gt;) is not null, it cancels it and stops listening to events from it.
		 */
		public void cancel(){
			if(subscription!=null){
				subscription.cancel();
				subscription.removeEventListener(MarketScanWorker.this);		
			}
		}
		/**
		 * returns whether this MarketScanWorker returned any results after the 15 sec timeout
		 * @return
		 */
		public boolean isResponding() {
			return isResponding.get();
		}
		

		
		public void eventFired(MarketScan liveMarketScan) {
			marketScanPendingFetch.offer(liveMarketScan);
//			liveMarketScans.add(0,liveMarketScan);

				
//			boolean saveRightAway=true;
//			if(saveRightAway){
//				ema.getMarketScanDao().update(liveMarketScans);
//			}else if(liveMarketScans.size()>CACHE_SIZE){
//				List<MarketScan> halfCache=liveMarketScans.subList(0, CACHE_SIZE/2);
//				List<MarketScan> marketScansToSave =liveMarketScans.subList((CACHE_SIZE/2)+1,liveMarketScans.size()-1);
//				if(RunMe.isRealRun&&useDatabaseForRealTimeMarketScans){
//					ema.getMarketScanDao().update(marketScansToSave);
//				}
//				liveMarketScans=halfCache;
//			}
			
//			if(scanHistory.size()>=HISTORY_SIZE)
//			{
//				scanHistory.poll();
//			}
//			scanHistory.offer(liveMarketScan);
			lastScanReceived=liveMarketScan.getTimeStamp();
			if(!isResponding.getAndSet(true)){
				synchronized(isResponding) {
					isResponding.notifyAll();
				}
			}
			if(processMarketScanChanges && marketScanEventProcessingStrategy!=null){//this is for marketScanChange events
				marketScanEventProcessingStrategy.processNewMarketScan(previousMarketScan, liveMarketScan);
			}
			previousMarketScan=liveMarketScan;
		}
		/**
		 * returns the associated liveMarketScans(List&lt;MarketScan&gt;)
		 * @return
		 */
		public List<MarketScan> getScanHistory() {
			return liveMarketScans;
		}
		/**
		 * returns the associated scanCriteria(ScanCriteria)
		 * @return
		 */
		public ScanCriteria getScanCriteria() {
			return scanCriteria;
		}		
	}

	/**
	 * returns the associated availableScanners(&lt;ScanCriteria, MarketScanWorker&gt;)<br/>
	 * These are the ones we know last time worked. It might be canceled to make room for the new ones, so we need to call
	 * activate to make them active again
	 * @return
	 */
	public Map<ScanCriteria, MarketScanWorker> getAvailableScanners() {
		return availableScanners;
	}
}
